{
  "metadata" : {
    "id" : "3a8c1b46-b256-4aec-9ee4-b85b9ad6f10f",
    "name" : "flatmapgroupswithstate-n-moving-average-timeout",
    "user_save_timestamp" : "2018-07-08T22:44:40.310Z",
    "auto_save_timestamp" : "1970-01-01T01:00:00.000Z",
    "language_info" : {
      "name" : "scala",
      "file_extension" : "scala",
      "codemirror_mode" : "text/x-scala"
    },
    "trusted" : true,
    "sparkNotebook" : null,
    "customLocalRepo" : null,
    "customRepos" : null,
    "customDeps" : null,
    "customImports" : null,
    "customArgs" : null,
    "customSparkConf" : null,
    "customVars" : null
  },
  "cells" : [ {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "9B48F85A440847599ABAC8D20BB1BE7A"
    },
    "cell_type" : "code",
    "source" : [ "import java.sql.Timestamp\n", "case class Rate(timestamp: Timestamp, value: Long)" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import java.sql.Timestamp\ndefined class Rate\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 1,
      "time" : "Took: 1.108s, at 2018-07-15 19:13"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "FAEC54572A50475CA75B4D518C82747F"
    },
    "cell_type" : "code",
    "source" : [ "// define some constants\n", "val ElementCountWindowSize = 10" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "ElementCountWindowSize: Int = 10\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 16,
      "time" : "Took: 0.461s, at 2018-07-15 19:17"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "96F16FB4DD5E48C68E6AD1CFDCB40CAD"
    },
    "cell_type" : "code",
    "source" : [ "val rate = sparkSession.readStream.format(\"rate\").load().as[Rate]" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "rate: org.apache.spark.sql.Dataset[Rate] = [timestamp: timestamp, value: bigint]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 2,
      "time" : "Took: 2.141s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "E041DCED2C0C4E788163C5F3C6B78E9F"
    },
    "cell_type" : "code",
    "source" : [ "val uids = List(\"d1e46a42\", \"d8e16e2a\", \"d1b06f88\", \n", "                \"d2e710aa\", \"d2f731cc\", \"d4c162ee\", \n", "                \"d4a11632\", \"d7e277b2\", \"d59018de\", \n", "                \"d60779f6\" )" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "uids: List[String] = List(d1e46a42, d8e16e2a, d1b06f88, d2e710aa, d2f731cc, d4c162ee, d4a11632, d7e277b2, d59018de, d60779f6)\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 3,
      "time" : "Took: 0.727s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "5F981F37D55349978CBC51F3A2EB7A01"
    },
    "cell_type" : "code",
    "source" : [ "val locationGenerator: () => (Double, Double) = {\n", "  // Europe bounds \n", "  val longBounds = (-10.89,39.82)\n", "  val latBounds = (35.52,56.7)\n", "  def pointInRange(bounds:(Double, Double)): Double = {\n", "    val (a, b) = bounds\n", "    Math.abs(scala.util.Random.nextDouble())*b+a\n", "  }\n", "  () => (pointInRange(longBounds), pointInRange(latBounds))\n", "}    " ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "locationGenerator: () => (Double, Double) = <function0>\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 4,
      "time" : "Took: 0.760s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "58404D4D92864DFD86BA95204F86FF42"
    },
    "cell_type" : "code",
    "source" : [ "def pickOne[T](list: List[T]): T = list(scala.util.Random.nextInt(list.size))" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "pickOne: [T](list: List[T])T\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 5,
      "time" : "Took: 0.610s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "02A8945CE374431F8F5269DCE7A8A4D3"
    },
    "cell_type" : "code",
    "source" : [ "val pressureGen: () => Double = () => scala.util.Random.nextDouble + 101.0\n", "val tempGen: () => Double = () => scala.util.Random.nextDouble * 60 - 20" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "pressureGen: () => Double = <function0>\ntempGen: () => Double = <function0>\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 6,
      "time" : "Took: 0.695s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "37D0E9DA64FC43D3A526B7386E09DF2D"
    },
    "cell_type" : "code",
    "source" : [ "import java.sql.Timestamp\n", "case class WeatherEvent(stationId: String, timestamp: Timestamp, location:(Double,Double), pressure: Double, temp: Double)" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import java.sql.Timestamp\ndefined class WeatherEvent\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 7,
      "time" : "Took: 0.515s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "5D9C60387781455DBC4AF900923E935B"
    },
    "cell_type" : "code",
    "source" : [ "val weatherEvents = rate.map{case Rate(ts, value) => WeatherEvent(pickOne(uids), ts, locationGenerator(), pressureGen(), tempGen())}" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "weatherEvents: org.apache.spark.sql.Dataset[WeatherEvent] = [stationId: string, timestamp: timestamp ... 3 more fields]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 8,
      "time" : "Took: 0.863s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "6E647F3B92404A5686145CE08F06AB07"
    },
    "cell_type" : "code",
    "source" : [ "import scala.collection.immutable.Queue\n", "case class FIFOBuffer[T](capacity: Int, data: Queue[T] = Queue.empty) extends Serializable {\n", "  def add(element: T): FIFOBuffer[T] = this.copy(data = data.enqueue(element).take(capacity))\n", "  def get: List[T] = data.toList\n", "  def size: Int = data.size\n", "}" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import scala.collection.immutable.Queue\ndefined class FIFOBuffer\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 9,
      "time" : "Took: 0.645s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "65092AC8EE20426D94919449A91B14A2"
    },
    "cell_type" : "code",
    "source" : [ "import java.sql.Timestamp\n", "case class WeatherEventAverage(stationId: String, \n", "                               startTime: Timestamp, \n", "                               endTime:Timestamp, \n", "                               pressureAvg: Double, \n", "                               tempAvg: Double)" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import java.sql.Timestamp\ndefined class WeatherEventAverage\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 10,
      "time" : "Took: 0.450s, at 2018-07-15 19:14"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "C9906C7DA0C141C9872A1B697AE773A6"
    },
    "cell_type" : "code",
    "source" : [ "def stateToAverageEvent(key: String, data: FIFOBuffer[WeatherEvent]):Iterator[WeatherEventAverage] = {\n", "  if (data.size == ElementCountWindowSize) {\n", "      val events = data.get\n", "      val start = events.head\n", "      val end = events.last\n", "      val pressureAvg = events.map(event => event.pressure).sum/data.size\n", "      val tempAvg = events.map(event => event.temp).sum/data.size\n", "      Iterator(WeatherEventAverage(key, start.timestamp, end.timestamp, pressureAvg, tempAvg))\n", "    } else {\n", "      Iterator.empty\n", "    }\n", "}" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import org.apache.spark.sql.streaming.GroupState\nstateToAverageEvent: (key: String, data: FIFOBuffer[WeatherEvent])Iterator[WeatherEventAverage]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 13,
      "time" : "Took: 0.756s, at 2018-07-15 19:15"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "DBAF0B87708D4CB5AA547B33B04F8863"
    },
    "cell_type" : "code",
    "source" : [ "import org.apache.spark.sql.streaming.GroupState\n", "def flatMappingFunction(key: String, values: Iterator[WeatherEvent], state: GroupState[FIFOBuffer[WeatherEvent]]): \n", "Iterator[WeatherEventAverage] = {\n", "  if (state.hasTimedOut) {\n", "    // when the state has a timeout, the values are empty\n", "    // this validation is only to illustrate that point\n", "    assert(values.isEmpty, \"When the state has a timeout, the values are empty\")\n", "    val result = stateToAverageEvent(key, state.get)\n", "    // evict the timed-out state\n", "    state.remove()\n", "    // emit the result of transforming the current state into an output record\n", "    result\n", "  } else {\n", "    // get current state or create a new one if there's no previous state\n", "    val currentState = state.getOption.getOrElse(new FIFOBuffer[WeatherEvent](ElementCountWindowSize))\n", "    // enrich the state with the new events\n", "    val updatedState = values.foldLeft(currentState){case (st, ev) => st.add(ev)}\n", "    // update the state with the enriched state\n", "    state.update(updatedState)\n", "    state.setTimeoutDuration(\"30 seconds\")\n", "    // only when we have enough data, create a WeatherEventAverage from the accumulated state\n", "    // before that, we return an empty result.\n", "    stateToAverageEvent(key, updatedState)\n", "  }\n", "}                                              " ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import org.apache.spark.sql.streaming.GroupState\nflatMappingFunction: (key: String, values: Iterator[WeatherEvent], state: org.apache.spark.sql.streaming.GroupState[FIFOBuffer[WeatherEvent]])Iterator[WeatherEventAverage]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 20,
      "time" : "Took: 0.509s, at 2018-07-15 19:38"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "7FF731E763BF4DBA8761BFB03CDC23B4"
    },
    "cell_type" : "code",
    "source" : [ "import org.apache.spark.sql.streaming.{GroupStateTimeout, OutputMode}\n", "val weatherEventsMovingAverage = weatherEvents\n", ".withWatermark(\"timestamp\", \"2 minutes\")\n", ".groupByKey(record => record.stationId)\n", ".flatMapGroupsWithState(OutputMode.Update, GroupStateTimeout.ProcessingTimeTimeout)(flatMappingFunction)" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "import org.apache.spark.sql.streaming.{GroupStateTimeout, OutputMode}\nweatherEventsMovingAverage: org.apache.spark.sql.Dataset[WeatherEventAverage] = [stationId: string, startTime: timestamp ... 3 more fields]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 21,
      "time" : "Took: 0.815s, at 2018-07-15 19:39"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "BAD771D079B6431286C956296D43BC66"
    },
    "cell_type" : "code",
    "source" : [ "val outQuery = weatherEventsMovingAverage.writeStream\n", "  .format(\"memory\")\n", "  .queryName(\"weatherAverage\")\n", "  .outputMode(\"update\")\n", "  .start()" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "outQuery: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@7ef6e0d5\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 22,
      "time" : "Took: 1.028s, at 2018-07-15 19:39"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "output_stream_collapsed" : true,
      "collapsed" : false,
      "id" : "7778099F1C3947A2873026A40087F405"
    },
    "cell_type" : "code",
    "source" : [ "outQuery.stop()" ],
    "outputs" : [ {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 20,
      "time" : "Took: 1.340s, at 2018-07-08 18:49"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "13482567EBD3442EA7BF629FADA559F3"
    },
    "cell_type" : "code",
    "source" : [ "val table  = sparkSession.sql(\"select * from weatherAverage\")" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "table: org.apache.spark.sql.DataFrame = [stationId: string, startTime: timestamp ... 3 more fields]\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 23,
      "time" : "Took: 1.053s, at 2018-07-15 19:39"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "1FBDD7607D83498C89FED6E46F7B9124"
    },
    "cell_type" : "code",
    "source" : [ "table.show(truncate= false)" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "+---------+-----------------------+-----------------------+------------------+------------------+\n|stationId|startTime              |endTime                |pressureAvg       |tempAvg           |\n+---------+-----------------------+-----------------------+------------------+------------------+\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d59018de |2018-07-15 19:39:41.681|2018-07-15 19:40:42.681|101.45304706385005|14.724579048505515|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d59018de |2018-07-15 19:39:41.681|2018-07-15 19:40:42.681|101.45304706385005|14.724579048505515|\n|d59018de |2018-07-15 19:39:41.681|2018-07-15 19:40:42.681|101.45304706385005|14.724579048505515|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d59018de |2018-07-15 19:39:41.681|2018-07-15 19:40:42.681|101.45304706385005|14.724579048505515|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d2e710aa |2018-07-15 19:39:51.681|2018-07-15 19:41:04.681|101.47805380773234|6.026307192727087 |\n|d2e710aa |2018-07-15 19:39:51.681|2018-07-15 19:41:04.681|101.47805380773234|6.026307192727087 |\n|d59018de |2018-07-15 19:39:41.681|2018-07-15 19:40:42.681|101.45304706385005|14.724579048505515|\n|d2e710aa |2018-07-15 19:39:51.681|2018-07-15 19:41:04.681|101.47805380773234|6.026307192727087 |\n|d1b06f88 |2018-07-15 19:39:38.681|2018-07-15 19:41:14.681|101.48914104278731|12.304328428450534|\n|d2e710aa |2018-07-15 19:39:51.681|2018-07-15 19:41:04.681|101.47805380773234|6.026307192727087 |\n|d4c162ee |2018-07-15 19:39:39.681|2018-07-15 19:40:08.681|101.55667628745744|-0.698490270514989|\n|d60779f6 |2018-07-15 19:40:04.681|2018-07-15 19:41:21.681|101.40461400392056|13.830357123999761|\n+---------+-----------------------+-----------------------+------------------+------------------+\nonly showing top 20 rows\n\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 25,
      "time" : "Took: 1.112s, at 2018-07-15 19:44"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "8CA3C13B054C4AFBAB9BC533EB2B0349"
    },
    "cell_type" : "code",
    "source" : [ "outQuery.stop" ],
    "outputs" : [ {
      "metadata" : { },
      "data" : {
        "text/html" : ""
      },
      "output_type" : "execute_result",
      "execution_count" : 17,
      "time" : "Took: 0.970s, at 2018-07-08 19:21"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : false,
      "id" : "2E81FDAE71174287860E457B81A1681E"
    },
    "cell_type" : "code",
    "source" : [ "outQuery.lastProgress" ],
    "outputs" : [ {
      "name" : "stdout",
      "output_type" : "stream",
      "text" : "res18: org.apache.spark.sql.streaming.StreamingQueryProgress =\n{\n  \"id\" : \"8fb47565-5bb7-4990-b8df-280767911647\",\n  \"runId\" : \"e425e9a5-1fb9-4706-8c99-ceaf3b7a0003\",\n  \"name\" : \"weatherAverage\",\n  \"timestamp\" : \"2018-07-08T20:58:58.535Z\",\n  \"numInputRows\" : 0,\n  \"inputRowsPerSecond\" : 0.0,\n  \"durationMs\" : {\n    \"getOffset\" : 0,\n    \"triggerExecution\" : 0\n  },\n  \"stateOperators\" : [ {\n    \"numRowsTotal\" : 10,\n    \"numRowsUpdated\" : 0\n  } ],\n  \"sources\" : [ {\n    \"description\" : \"RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8]\",\n    \"startOffset\" : 60,\n    \"endOffset\" : 60,\n    \"numInputRows\" : 0,\n    \"inputRowsPerSecond\" : 0.0\n  } ],\n  \"sink\" : {\n    \"description\" : \"MemorySink\"\n  }\n}\n"
    }, {
      "metadata" : { },
      "data" : {
        "text/html" : "{\n  &quot;id&quot; : &quot;8fb47565-5bb7-4990-b8df-280767911647&quot;,\n  &quot;runId&quot; : &quot;e425e9a5-1fb9-4706-8c99-ceaf3b7a0003&quot;,\n  &quot;name&quot; : &quot;weatherAverage&quot;,\n  &quot;timestamp&quot; : &quot;2018-07-08T20:58:58.535Z&quot;,\n  &quot;numInputRows&quot; : 0,\n  &quot;inputRowsPerSecond&quot; : 0.0,\n  &quot;durationMs&quot; : {\n    &quot;getOffset&quot; : 0,\n    &quot;triggerExecution&quot; : 0\n  },\n  &quot;stateOperators&quot; : [ {\n    &quot;numRowsTotal&quot; : 10,\n    &quot;numRowsUpdated&quot; : 0\n  } ],\n  &quot;sources&quot; : [ {\n    &quot;description&quot; : &quot;RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8]&quot;,\n    &quot;startOffset&quot; : 60,\n    &quot;endOffset&quot; : 60,\n    &quot;numInputRows&quot; : 0,\n    &quot;inputRowsPerSecond&quot; : 0.0\n  } ],\n  &quot;sink&quot; : {\n    &quot;description&quot; : &quot;MemorySink&quot;\n  }\n}"
      },
      "output_type" : "execute_result",
      "execution_count" : 17,
      "time" : "Took: 0.799s, at 2018-07-08 22:59"
    } ]
  }, {
    "metadata" : {
      "trusted" : true,
      "input_collapsed" : false,
      "collapsed" : true,
      "id" : "AD0F92870C884C5084A796F5DCD68F9C"
    },
    "cell_type" : "code",
    "source" : [ "" ],
    "outputs" : [ ]
  } ],
  "nbformat" : 4
}